package com.cssl.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/11 20:00
 * @Desc:
 */
public class CommonUtils {
    private static Properties properties = new Properties();

    static {
        try {
            properties.load(new InputStreamReader(Objects.requireNonNull(Thread.currentThread().getContextClassLoader().getResourceAsStream("config.properties")), StandardCharsets.UTF_8));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Properties getProperties() {
        return properties;
    }

    public static <T> FlinkKafkaProducer<T> getKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        String brokerList = properties.getProperty("kafka.broker.list");

        Properties kafkaProperties = new Properties();
        kafkaProperties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        String defaultTopic = properties.getProperty("kafka.default.topic");
        return new FlinkKafkaProducer<T>(
                defaultTopic,
                kafkaSerializationSchema,
                kafkaProperties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) throws IOException {
        String brokerList = properties.getProperty("kafka.broker.list");
        return new FlinkKafkaProducer<String>(
                brokerList,
                topic,
                new SimpleStringSchema());
    }

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic) {
        Properties kafkaProperties = new Properties();
        kafkaProperties.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getProperty("kafka.consumer.group"));
        kafkaProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getProperty("kafka.broker.list"));
        return new FlinkKafkaConsumer<String>(
                topic,
                new SimpleStringSchema(),
                kafkaProperties);
    }

    public static String getKafkaDDL(String sourceTopic, String groupId) {
        String brokerList = properties.getProperty("kafka.broker.list");

        return "'connector' = 'kafka', " +
                " 'topic' = '" + sourceTopic + "'," +
                " 'properties.bootstrap.servers' = '" + brokerList + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'latest-offset' ";
    }
}
